package realtime.app.dwd.db;/*
 *@ClassName : dwd_test_exam
 *@Description : $ {description}
 *@Author : NingYu
 *@Date : 2023-03-24 16:{MINUTE}
 */

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import realtime.util.MyKafkaUtil;


public class dwd_test_exam {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_test_exam_02"));

        Table testExamTable = tableEnv.sqlQuery("" +
                "select\n" +
                "    data['id'] id,\n" +
                "    data['paper_id'] paper_id,\n" +
                "    data['user_id'] user_id,\n" +
                "    cast(data['score'] as decimal(16,2)) score,\n" +
                "    cast(data['duration_sec'] as decimal(16,2)) duration_sec,\n" +
                "    date_format(data['create_time'],'yyyy-MM-dd') create_date,"+
                "    data['create_time'] create_time,\n" +
                "    data['submit_time'] submit_time,\n" +
                "    data['update_time'] update_time,\n" +
                "    data['deleted'] deleted,\n" +
                "    ts\n" +
                "from topic_db\n" +
                "where `table` = 'test_exam'\n" +
                "    and `type` = 'insert'\n" +
                "    and data['deleted']='0'");


        tableEnv.createTemporaryView("test_exam",testExamTable);

        tableEnv.executeSql("create table dwd_test_exam(\n" +
                "    id String,\n" +
                "    paper_id String,\n" +
                "    user_id String,\n" +
                "    score decimal(16,2),\n" +
                "    duration_sec decimal(16,2),\n" +
                "    create_date String,"+
                "    create_time String,\n" +
                "    submit_time String,\n" +
                "    update_time String,\n" +
                "    `deleted` String,\n" +
                "    ts BIGINT,\n" +
                "primary key(id) not enforced" +
                ")"+MyKafkaUtil.getUpsertKafkaDDL("dwd_test_exam")

        );

        tableEnv.executeSql("insert into dwd_test_exam select * from test_exam");


    }
}
